package com.goodwill.hdr.dao;

import com.goodwill.hdr.Strategy.HbaseFilterStrategy;
import com.goodwill.hdr.vo.PropertyFilter;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;

/**
 * @version V1.0
 * @Description:
 * @author: yanhengtao
 * @date: 2020年3月10日
 */
@Service
public class ConsoleQueryDao {

    private Logger logger = LoggerFactory.getLogger(getClass());


    private final Map<String, HbaseFilterStrategy> hbaseFilterStrategyMap;

    private final Connection connection;
    @Value("${patientId.start}")
    private String configPatientIdStart;
    @Value("${patientId.end}")
    private String configPatientIdEnd;


    public ConsoleQueryDao(@Qualifier("hbaseFilterStrategyMap") Map<String, HbaseFilterStrategy> hbaseFilterStrategyMap, Connection connection) {

        this.hbaseFilterStrategyMap = hbaseFilterStrategyMap;
        this.connection = connection;
    }





    public List<Map<String, String>> queryHBaseTableData(String tableName, String patientId,String oid,
                                                         List<PropertyFilter> filters, String... columns) {
        logger.info("[{开始查询hbase表 " + tableName + "}]");
        logger.info("[{patientId: " + patientId+ "}]");
        Scan scan = new Scan();


        addFilterList(filters, scan);
        String rowKey = getRowKey(oid, patientId);
        byte[] startRow = rowKey.getBytes(StandardCharsets.UTF_8);
        byte[] stopRow = (rowKey + "}").getBytes(StandardCharsets.UTF_8);
        scan.withStartRow(startRow);
        scan.withStopRow(stopRow);

        if (columns.length > 0) {
            for (String column : columns) {
                scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes(column));
            }
        }else {
            scan.addFamily(Bytes.toBytes("cf"));
        }
        ResultScanner scanner = null;
        Table table = null;
        List<Map<String, String>> queryResult = new ArrayList<>();
        try {

            table = connection.getTable(TableName.valueOf(tableName));
            scanner = table.getScanner(scan);
            for (Result result : scanner) {
                NavigableMap<byte[], NavigableMap<byte[], byte[]>> noVersionMap = result.getNoVersionMap();
                NavigableMap<byte[], byte[]> map = noVersionMap.get(Bytes.toBytes("cf"));
                Map<String, String> stringMap = byteMapToStringMap(map);
                stringMap.put("rowKey",new String(result.getRow()));
                queryResult.add(stringMap);
            }
        } catch (IOException e) {
            throw new RuntimeException("查询hbase发送IO异常", e);
        } finally {
            if (scanner != null) {
                scanner.close();
            }
            if (table != null) {
                try {
                    table.close();
                } catch (IOException e) {
                    throw new RuntimeException("hbase table 关闭失败", e);
                }
            }
        }

        logger.info("hbase表 " + tableName + "最终返回数据量为{}",queryResult.size());
        return queryResult;
    }




    private Map<String, String> byteMapToStringMap(Map<byte[], byte[]> source) {
        Map<String, String> result = new HashMap<>();
        for (byte[] key : source.keySet()) {
            String resultKey = Bytes.toString(key);
            String resultValue = Bytes.toString(source.get(key));
            result.put(resultKey, resultValue);
        }
        return result;
    }

    /**
     * 返回过滤条件集合
     */
    public void addFilterList(List<PropertyFilter> propertyFilterList, Scan scan) {

        if (propertyFilterList != null && !propertyFilterList.isEmpty()) {
            FilterList filterList = new FilterList();
            for (PropertyFilter filter : propertyFilterList) {
                String matchType = filter.getMatchType();
                String column = filter.getPropertyName();
                HbaseFilterStrategy filterStrategy = hbaseFilterStrategyMap.get(matchType);
                scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes(column));
                filterStrategy.execute(filter, filterList);
            }
            scan.setFilter(filterList);
        }

    }

    /**
     * 获取rowkey
     *
     * @param searchValue 查询值
     * @return
     */
    public String getRowKey(String oid, String searchValue) {
        String rowkey = "";


        rowkey = getRowkeyPrefix(searchValue, oid);

        return rowkey;
    }

    private String getRowkeyPrefix(String patient_id, String oid) {
        String salt = getSalt(patient_id);
        return salt + "|" + oid + "|" + patient_id;
    }

    private String getSalt(String patient_id) {
        String salt = "";

        Integer patient_id_start = Integer.parseInt(configPatientIdStart);


        Integer patient_id_end = Integer.parseInt(configPatientIdEnd);


        if (patient_id_end - patient_id_start < 3) {
            return "zzzz";
        }


        StringBuffer sb = new StringBuffer(patient_id.substring(
                patient_id.length() + patient_id_start, patient_id.length() +1+ patient_id_end));
        salt = sb.reverse().toString();
        return salt;
    }


}
